package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.udf.TimeTruncate;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/6
 * @Desc: 学大数据，到多易教育
 *  流量概况多维分析： 按用户、分钟粒度 聚合pv数
 **/
public class Etl02_TrafficOverviewAggByUserMinute {
    public static void main(String[] args) {
        // 创建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        env.setStateBackend(new HashMapStateBackend());
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建kafka连接器表（字段+连接器参数），映射kafka中dwd层的公共维度打宽数据topic
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka (                           "
                        +"     user_id           BIGINT,                     "
                        +"     username          string,                     "
                        +"     session_id        string,                     "
                        +"     event_id          string,                     "
                        +"     event_time        bigint,                     "
                        +"     lat               double,                     "
                        +"     lng               double,                     "
                        +"     release_channel   string,                     "
                        +"     device_type       string,                     "
                        +"     properties        map<string,string>,         "
                        +"     register_phone    STRING,                     "
                        +"     user_status       INT,                        "
                        +"     register_time     TIMESTAMP(3),               "
                        +"     register_gender   INT,                        "
                        +"     register_birthday DATE,                       "
                        +"     register_province STRING,                     "
                        +"     register_city        STRING,                  "
                        +"     register_job         STRING,                  "
                        +"     register_source_type INT,                     "
                        +"     gps_province STRING,                          "
                        +"     gps_city     STRING,                          "
                        +"     gps_region   STRING,                          "
                        +"     page_type    STRING,                          "
                        +"     page_service STRING,                          "
                        +"     rt AS to_timestamp_ltz(event_time,3),         "
                        +"     WATERMARK FOR rt AS rt - interval '0' second  "
                        +" ) WITH (                                          "
                        +"  'connector' = 'kafka',                           "
                        +"  'topic' = 'dwd-events-detail',                   "
                        +"  'properties.bootstrap.servers' = 'doitedu:9092', "
                        +"  'properties.group.id' = 'testGroup',             "
                        +"  'scan.startup.mode' = 'earliest-offset',         "
                        +"  'value.format'='json',                           "
                        +"  'value.json.fail-on-missing-field'='false',      "
                        +"  'value.fields-include' = 'EXCEPT_KEY')           ");


        // 创建doris连接器表，映射轻度聚合后的目标存储表：dws.tfc_overview_u_m
        tenv.executeSql(
                " CREATE TABLE tfc_overview_u_m_doris(         "
                        +" dt  DATE,                 "
                        +" time_60m STRING,          "
                        +" time_30m STRING,           "
                        +" time_10m STRING,          "
                        +" time_m STRING,            "
                        +" user_id BIGINT,           "
                        +" is_newuser INT,           "
                        +" session_id STRING,        "
                        +" release_channel STRING,   "
                        +" device_type STRING,       "
                        +" gps_province STRING,      "
                        +" gps_city STRING,          "
                        +" gps_region STRING,        "
                        +" page_type STRING,         "
                        +" page_service STRING,      "
                        +" page_url STRING,          "
                        +" pv_amt BIGINT             "
                        + " ) WITH (                               "
                        + "    'connector' = 'doris',              "
                        + "    'fenodes' = 'doitedu:8030',         "
                        + "    'table.identifier' = 'dws.tfc_overview_u_m',  "
                        + "    'username' = 'root',                "
                        + "    'password' = '',                    "
                        + "    'sink.label-prefix' = 'doris_label_dws"+System.currentTimeMillis()+"'"
                        + " )                                         ");

        // 开发一个聚合sql
        tenv.createTemporaryFunction("time_truncate", TimeTruncate.class);
        tenv.executeSql(
                " INSERT INTO  tfc_overview_u_m_doris                                                    "
                        +" WITH tmp AS (                                                                    "
                        +" SELECT                                                                           "
                        +"     user_id,                                                                     "
                        +"     IF(DATE_FORMAT(register_time,'yyyy-MM-dd')<CURRENT_DATE,0,1) AS is_newuser,  "
                        +"     session_id,                                                                  "
                        +"     release_channel,                                                             "
                        +"     device_type,                                                                 "
                        +"     gps_province,                                                                "
                        +"     gps_city,                                                                    "
                        +"     gps_region,                                                                  "
                        +"     page_type,                                                                   "
                        +"     page_service,                                                                "
                        +"     properties['url'] AS page_url,                                               "
                        +"     rt                                                                           "
                        +" FROM dwd_kafka                                                                   "
                        +" WHERE event_id = 'page_load'                                                     "
                        +" )                                                                                "
                        +" SELECT                                                                           "
                        +"     TO_DATE(DATE_FORMAT(window_start,'yyyy-MM-dd')) AS   dt,                     "
                        +"     time_truncate(CAST(window_start AS STRING),60) AS time_60m,                     "
                        +"     time_truncate(CAST(window_start AS STRING),30) AS time_30m,                     "
                        +"     time_truncate(CAST(window_start AS STRING),10) AS time_10m,                     "
                        +"     CAST(window_start AS STRING) AS time_m,                                      "
                        +"     user_id,                                                                     "
                        +"     is_newuser,                                                                  "
                        +"     session_id,                                                                  "
                        +"     release_channel,                                                             "
                        +"     device_type,                                                                 "
                        +"     gps_province,                                                                "
                        +"     gps_city,                                                                    "
                        +"     gps_region,                                                                  "
                        +"     page_type,                                                                   "
                        +"     page_service,                                                                "
                        +"     page_url,                                                                    "
                        +"     COUNT(1) AS pv_amt                                                           "
                        +" FROM TABLE(                                                                      "
                        +"     TUMBLE(TABLE tmp, DESCRIPTOR(rt), INTERVAL '1' MINUTE)                     "
                        +" )                                                                                "
                        +" GROUP BY                                                                         "
                        +"     window_start,                                                                "
                        +"     window_end,                                                                  "
                        +"     user_id,                                                                     "
                        +"     is_newuser,                                                                  "
                        +"     session_id,                                                                  "
                        +"     release_channel,                                                             "
                        +"     device_type,                                                                 "
                        +"     gps_province,                                                                "
                        +"     gps_city,                                                                    "
                        +"     gps_region,                                                                  "
                        +"     page_type,                                                                   "
                        +"     page_service,                                                                "
                        +"     page_url                                                                     "
        );
    }
}
